/*********************************************************
          File Name:pthread_barrier.c
          Author: Abby Cin
          Mail: abbytsing@gmail.com
          Created Time: Sat 29 Aug 2015 04:59:51 PM CST
**********************************************************/

#include<stdio.h>
#include<pthread.h>
#include<unistd.h>
#include<stdlib.h>
#include<errno.h>
#include<fcntl.h>
#include<sys/stat.h>
#include<sys/types.h>

char *file_in = NULL;
char *file_out = NULL;

static void err(const char *s){
    perror(s);
    exit(errno);
}

typedef struct{
    char part[10];
    long block;
    long start;
} pkg;

pthread_barrier_t ctx;
pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;

int pwreadn(int fdin, int fdout, long offt, long size)
{
    long len = size;
    long res = 0;
    char *buf = (char *)malloc(size);

    while(len)
    {
        res = read(fdin, buf, len);

        if(res == -1)
            return -1;
        if(res == 0)
            break;
        len -= res;

        if(pwrite(fdout, buf, res, offt) == -1)
            return -1;
    }
    return 0;
}

void merge(pkg *rpm[])
{
    int fin, fout, i;
    char *buf;

    buf = (char *)malloc(rpm[0]->block);

    fout = open(file_out, O_CREAT|O_RDWR, 00644);
    if(fout == -1)
        err("open target file");
    for(i = 0; i<4; i++)
    {
        fin=open(rpm[i]->part, O_RDWR);
        if(fin == -1) err("open src file");
        
        while(read(fin, buf, rpm[i]->block))
            write(fout, buf, rpm[i]->block);
        close(fin);
        unlink(rpm[i]->part);
        printf("++++++++++++ merge\e[36m %s \e[0mdone ++++++++++\n", rpm[i]->part);
    }
    close(fout);
}

void *pfun(void *tmp)
{
    pkg *pc = (pkg *)tmp;

    int fdin, fdout;
    char res = 0;
    fdin = open(file_in, O_RDWR);
    fdout = open(pc->part, O_CREAT|O_RDWR, 00644);

    if(fdin == -1 || fdout == -1)
        err("open");
    lseek(fdin, pc->start, SEEK_SET);

    printf("start = %ld\t block = %ld\n", pc->start, pc->block);
    pthread_mutex_lock(&mtx);

    printf("thread: %lu\n", pthread_self());

    res = pwreadn(fdin, fdout, 0, pc->block);
    if(res == -1)
        printf("thread: %lu --> rd-wr failed\n", pthread_self());
    close(fdin);
    close(fdout);

    pthread_mutex_unlock(&mtx);

    pthread_barrier_wait(&ctx); /* necessary */
    pthread_exit(&res);
}

int main(int argc, char *argv[])
{
    if(argc != 3)
    {
        fprintf(stderr, "%s srcfile  destfile\n", argv[0]);
        exit(1);
    }
    struct stat st;
    pkg *rpm[4];
    long size,tmp,tmp1;
    int i;
    pthread_t tid[4];

    for(i=0; i<4; i++){
        rpm[i] = (pkg *)malloc(sizeof(pkg));
        rpm[i]->start = 0;
        rpm[i]->block = 0;
    }

    stat(argv[1], &st);

    size = st.st_size;
    tmp = size / 4;
    tmp1  = size % 4;

    file_in = argv[1];
    file_out = argv[2];

    pthread_barrier_init(&ctx, NULL, 5);
    for(i = 0; i< 4; i++)
    {
        rpm[i]->block = i == 0 ? tmp + tmp1 : tmp;
        rpm[i]->start = i == 0 ? 0 : (i == 1 ? tmp + tmp1 : rpm[i-1]->start + tmp);
        sprintf(rpm[i]->part, "part%d", i);
        if(pthread_create(&tid[i], NULL, pfun, (void *)rpm[i]) != 0)
            break;
    }
     
    pthread_barrier_wait(&ctx);
    printf("\e[37m child threads done, start merging !\e[0m\n");
    merge(rpm);
    
    for(i = 0; i<4; i++)
        if(pthread_join(tid[i], (void *)NULL) != 0)
            fprintf(stderr, "join failed\n");
    pthread_barrier_destroy(&ctx);
    for(i=0; i<4; i++)
        free(rpm[i]);
    return errno;
}
